package cas.ihep.hadoop.rand;

import cas.ihep.fs.UnifiedFile;
import cas.ihep.fs.UnifiedFileSystem;
import com.google.common.io.Closer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class RandMapper extends Mapper<LongWritable, Text, Text, Text>  {

    static{
        if(System.getProperty("alluxio.site.conf.dir")==null){
            String alluxio_home=System.getenv("ALLUXIO_HOME");
            alluxio_home=alluxio_home+"/conf";
            System.setProperty("alluxio.site.conf.dir",alluxio_home);
        }
    }

    protected void map(LongWritable key, Text value, Context context) throws IOException{
        try(Closer closer=Closer.create()) {
            UnifiedFileSystem ufs = closer.register(UnifiedFileSystem.get(UnifiedFileSystem.newConfiguration(context.getConfiguration())));
            String line=value.toString();
            int idx=line.indexOf(' ');
            String fname=line.substring(0,idx);
            int origin=Integer.parseInt(line.substring(idx+1));
            byte[] buf=new byte[4096];
            FileInputStream fis = closer.register(new FileInputStream("/dev/urandom"));
            for(int i=0;i<5;i++) {
                UnifiedFile file = closer.register(ufs.create(new URI("alluxio://hadoop06.ihep.ac.cn:19998/output/" + fname + ".txt."+i)));
                int length = origin;
                FileOutputStream fos = closer.register(new FileOutputStream(file.next().absolutePath()));
                while (length > 0) {
                    int each = fis.read(buf, 0, Math.min(4096, length));
                    fos.write(buf, 0, each);
                    length -= each;
                }
            }
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }
}
